package com.small.nacos.core.cluster;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.common.utils.ConcurrentHashSet;
import com.small.nacos.common.notify.Event;
import com.small.nacos.common.notify.NotifyCenter;
import com.small.nacos.core.cluster.lookup.LookupFactory;
import com.small.nacos.sys.env.EnvUtil;
import com.small.nacos.sys.utils.InetUtils;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author zhoujin
 * @Date 2022/10/31 20:00
 */
@Component(value = "serverMemberManager")
public class ServerMemberManager {

    private volatile ConcurrentSkipListMap<String, Member> serverList;

    /**
     * 当前节点端口号
     */
    private int port;

    /**
     * 当前节点请求地址
     */
    private String localAddress;

    /**
     * 当前节点
     */
    private volatile Member self;

    /**
     * 成员查找
     */
    private MemberLookup lookup;


    private volatile Set<String> memberAddressInfos = new ConcurrentHashSet<>();

    public ServerMemberManager() throws Exception {
        this.serverList = new ConcurrentSkipListMap();
        init();
    }

    public void init() throws NacosException {
        this.port = EnvUtil.getProperty("server.port", Integer.class, 8848);
        this.localAddress = InetUtils.getSelfIP() + ":" + port;
        this.self = MemberUtil.singleParse(this.localAddress);
        serverList.put(self.getAddress(), self);

        //初始化服务改变事件
        registerClusterEvent();

        initAndStartLookup();

        if (serverList.isEmpty()) {
            throw new NacosException(NacosException.SERVER_ERROR, "cannot get serverlist, so exit.");
        }


    }

    private void initAndStartLookup() throws NacosException {
        this.lookup = LookupFactory.createLookUp(this);
        this.lookup.start();
    }

    private void registerClusterEvent() {
        // Register node change events
        NotifyCenter.registerToPublisher(MembersChangeEvent.class,
                EnvUtil.getProperty("nacos.member-change-event.queue.size", Integer.class, 128));
    }


    /**
     * 合并成员
     *
     * @param members
     * @return
     */
    synchronized boolean memberChange(Collection<Member> members) {
        if (members == null || members.size() == 0) {
            return false;
        }

        boolean isContainSelfIp = members.stream().anyMatch(selfMember -> Objects.equals(localAddress, selfMember.getAddress()));
        if (!isContainSelfIp) {
            members.add(this.self);
        }

        boolean hasChange = members.size() != serverList.size();
        ConcurrentSkipListMap<String, Member> tmpMap = new ConcurrentSkipListMap<>();
        Set<String> tmpAddressInfo = new ConcurrentHashSet<>();

        for (Member member : members) {

            final String address = member.getAddress();
            if (!serverList.containsKey(address)) {
                hasChange = true;

                member.setState(NodeState.DOWN);
            } else {
                member.setState(serverList.get(address).getState());
            }
            tmpMap.put(member.getAddress(), member);
            if (NodeState.UP.equals(member.getState())) {
                tmpAddressInfo.add(address);
            }
        }

        serverList = tmpMap;
        memberAddressInfos = tmpAddressInfo;

        //同步数据到文件中,这里会将原来的serverList跟换
        if (hasChange) {
            MemberUtil.syncToFile(allMembers());
            Event event = MembersChangeEvent.builder().members(allMembers()).build();
            NotifyCenter.publishEvent(event);
        }
        return hasChange;
    }

    public Collection<Member> allMembers() {
        HashSet<Member> set = new HashSet<>(serverList.values());
        set.add(self);
        return set;
    }
}
